package ma.zhiqiang.dataline.connector.sink;

import cn.hutool.core.util.ClassUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import ma.zhiqiang.dataline.core.dto.ColumnMap;
import ma.zhiqiang.dataline.core.engine.Sink;
import ma.zhiqiang.dataline.core.dto.SinkRecord;
import ma.zhiqiang.dataline.core.format.WriterFormatCsv;
import ma.zhiqiang.dataline.core.format.WriterFormatJson;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class KafkaSink implements Sink {
    Properties properties;
    KafkaTemplate kafkaTemplate;
    String topic;
    String format;

    public KafkaSink(Properties properties) {
        this.properties = properties;
        Map<String, Object> props = new HashMap<>(properties.size());
        properties.keySet().forEach(key->{
            props.put(key.toString(),properties.get(key));
        });
        DefaultKafkaProducerFactory defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(props);
        kafkaTemplate = new KafkaTemplate<>(defaultKafkaProducerFactory);
        topic = properties.getProperty("topic");
        format = properties.getProperty("format","text");
    }

    @Override
    public Boolean writer(SinkRecord sinkRecord) {
        String outline = "";
        if("csv".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatCsv().format(sinkRecord);
        }else if("json".equals(format) && (sinkRecord.getData() instanceof ColumnMap)){
            outline = new WriterFormatJson().format(sinkRecord);
        }else if(sinkRecord.getData() instanceof String){  //"text".equals(format)
            outline = sinkRecord.getData().toString();
        }else if(sinkRecord.getData() instanceof ColumnMap){
            outline = new WriterFormatJson().format(sinkRecord);
        }
        ListenableFuture<SendResult> future =kafkaTemplate.send(topic,outline);
        log.info("topic = {},message = {}",topic ,outline);
        AtomicReference<Boolean> bool = new AtomicReference<>(Boolean.TRUE);
        future.addCallback((sendResult)->{
            bool.set(Boolean.TRUE);
        },(throwable)->{
            log.error("发送kafka失败:",throwable);
            bool.set(Boolean.FALSE);
        });
        return bool.get();
    }
}
